package com.pengheng.sink.common;

import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.io.Serializable;
import java.sql.PreparedStatement;
import java.sql.SQLException;

public class CommonJdbcSink<T> implements Serializable {

    private String url = "jdbc:mysql://192.168.160.16:3306/gov2x?serverTimezone=Asia/Shanghai&useUnicode=true&characterEncoding=UTF-8";
    private String username = "root";
    private String password = "v2x@admin123";

    public SinkFunction<T> extracted(String sql, JdbcSupplier<T> supplier) {
        return JdbcSink.sink(sql, new JdbcStatementBuilder<T>() {
            @Override
            public void accept(PreparedStatement preparedStatement, T t) throws SQLException {
//                supplier.get(preparedStatement, t);

            }
        }, JdbcExecutionOptions.builder().withMaxRetries(3).withBatchSize(100).withBatchIntervalMs(3000).build(), new JdbcConnectionOptions.JdbcConnectionOptionsBuilder().withUrl(url).withUsername(username).withPassword(password).withConnectionCheckTimeoutSeconds(60).build());
    }
}
